package com.nulldev.util.concurrency.threadIt.v4.executors;

import java.util.ArrayList;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ft.FutureTask;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TEConditions;
import com.nulldev.util.data.Arrays.arrays.FastArrayList;
import com.nulldev.util.data.FastMath.FastMath;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

/**
 * The PBTE (Pulse [Interval] Based Thread Executor) executor. <br>
 * <br>
 * The PBTE schedules tasks based upon three intervals, ACTIVE, IDLE and
 * FULLY_IDLE. <br>
 * ACTIVE is used when <code>tasks > 4</code>, defaulting to every 20ms.<br>
 * IDLE is used when <code>tasks <= 4 && tasks > 1</code>, defaulting to every
 * 85ms.<br>
 * FULLY_IDLE is used when <code>tasks <= 1</code>, defaulting to every 150ms.
 * <br>
 * <br>
 * <b style="color: orange">This executor is highly experimental!</b>
 * 
 * @author nullifed
 */
public class PBTE extends AbstractExecutorService implements Runnable, ExtendedExecutor {
	public static final long DEFAULT_ACTIVE_PULSE_INTERVAL = 20;
	public static final long DEFAULT_IDLE_PULSE_INTERVAL = 85;
	public static final long DEFAULT_FULLY_IDLE_PULSE_INTERVAL = 240;
	public static final long DEFAULT_COOLDOWN_PEROID = 235;
	public static final int DEFAULT_THREAD_COUNT = 48;

	private final LinkedTransferQueue<Executable<?>> tasks = new LinkedTransferQueue<Executable<?>>();
	private final ArrayList<PBTEThread> executors = new FastArrayList<PBTEThread>(true);
	private final static Logger log = LoggerFactory.getLogger();

	public static enum PBTEThreadState {
		IDLE,
		BUSY,
		STOPPED
	}

	@SuppressWarnings("unused")
	private abstract class PBTEThread extends Thread implements Runnable {

		private volatile Executable<?> task;
		private final AtomicBoolean localSigStop = new AtomicBoolean();

		public PBTEThread() {
			super("PBTEThread");
		}

		public PBTEThread(final int threadID) {
			super("PBTEThread: #" + threadID);
		}

		@Override
		public void run() {
			while (!shouldStop() && !this.localSigStop.get()) {
				if (this.task != null) {
					this.state.lazySet(PBTEThreadState.BUSY);
					try {
						this.task.run(this);
						// log.info("ran: " + this.task);
						try {
							Thread.sleep(PBTE.this.getCooldownInterval());
						} catch (Exception ex) {
						}
					} catch (Throwable e) {
						log.error(e);
					}
					this.task = null;
					this.state.lazySet(PBTEThreadState.IDLE);
					try {
						Thread.sleep(PBTE.this.activePulseInterval);
					} catch (Exception e) {
					}
				} else {
					try {
						this.task = null;
						this.state.lazySet(PBTEThreadState.IDLE);
						Thread.sleep(PBTE.this.getPBTESleepAmount());
					} catch (Exception e) {
					}
				}
			}
		}

		public boolean setTask(final Executable<?> t) {
			if (this.task != null) {
				return false;
			} else {
				this.task = t;
				return true;
			}
		}

		public boolean hasTask() {
			return this.task != null;
		}

		public abstract boolean shouldStop();

		private final AtomicReference<PBTEThreadState> state = new AtomicReference<PBTEThreadState>(PBTEThreadState.IDLE);

		public PBTEThreadState getThreadState() {
			return this.state.get();
		}

		public boolean isBusy() {
			return this.state.get() == PBTEThreadState.BUSY || this.state.get() == PBTEThreadState.STOPPED;
		}

		public boolean isStopped() {
			return this.state.get() == PBTEThreadState.STOPPED;
		}

		public boolean stopThread() {
			if (this.isBusy())
				return false;
			if (this.isDaemon()) {
				this.setDaemon(false);
			}
			this.localSigStop.set(true);
			while (!this.isInterrupted())
				this.interrupt();
			this.state.set(PBTEThreadState.STOPPED);
			return true;
		}
	}

	private final int threadLimit;
	private final AtomicBoolean sig_stop = new AtomicBoolean();

	private long getSleepAmount() {
		final int size = this.tasks.size();
		if (size == 0) {
			return this.fullIdlePulseInterval;
		} else if (size == 1) {
			return this.almostIdlePulseInterval;
		} else if (size <= 6) {
			return this.idlePulseInterval + 40;
		} else if (size <= 10) {
			return this.idlePulseInterval;
		} else {
			return this.activePulseInterval;
		}
	}

	private long getPBTESleepAmount() {
		final int size = this.tasks.size();
		if (size <= 8) {
			return this.fullIdlePulseInterval;
		} else {
			return this.idlePulseInterval;
		}
	}

	private long getCooldownInterval() {
		final int size = this.tasks.size();
		if (size == 0) {
			return this.cooldownInterval;
		} else if (size <= 2) {
			return this.cooldownInterval / 2;
		} else if (size <= 8) {
			return this.cooldownInterval / 8;
		} else {
			return this.cooldownInterval / size;
		}
	}

	public PBTE() {
		this(DEFAULT_THREAD_COUNT);
	}

	public PBTE(final int threadCount) {
		this(threadCount, false);
	}

	private final Thread pbteController = new Thread(PBTE.this, "PBTE Executor: Dispatcher");

	public PBTE(final int threadCount, final boolean high_priority) {
		this.threadLimit = threadCount;
		this.__init_threads(high_priority);
		this.pbteController.start();
	}

	@Override
	public void run() {
		while (!this.sig_stop.get()) {
			try {
//				if (this.tasks.isEmpty()) {
//					Thread.sleep(this.idlePulseInterval);
//					continue;
//				}else {
				// log.info("tasks: " + this.tasks.size());
				final Executable<?> o = this.tasks.take();
				boolean success = false;
				for (final PBTEThread pbte : this.executors) {
					if (pbte.isBusy() || pbte.hasTask()) {
						// log.warn("PBTE::Thread is busy, skipped.");
						continue;
					} else {
						try {
							pbte.interrupt();
							pbte.setTask(o);
							success = true;
							break;
						} catch (NoSuchElementException e) {
						}
					}
				}
				if (!success) {
					this.tasks.add(o);
				}
//					Thread.sleep(this.getSleepAmount());
//				}
			} catch (Exception e) {
				log.error(e);
				try {
					Thread.sleep(this.getSleepAmount());
				} catch (InterruptedException e1) {
				}
			}
		}
	}

	private void __init_threads(final boolean high_priority) {
		if (high_priority) {
			log.warn("__init_threads(bool) -> High priority execution enabled! This may cause system hangups!");
		}
		for (int i = 0; i < threadLimit; i++) {
			final PBTEThread pbte = new PBTEThread(i) {

				@Override
				public boolean shouldStop() {
					return PBTE.this.sig_stop.get();
				}
			};
			if (high_priority) {
				pbte.setPriority(Thread.MAX_PRIORITY);
			}
			pbte.start();
			this.executors.add(pbte);
		}
	}

	private volatile long activePulseInterval = DEFAULT_ACTIVE_PULSE_INTERVAL;
	private volatile long idlePulseInterval = DEFAULT_IDLE_PULSE_INTERVAL;
	private volatile long almostIdlePulseInterval = DEFAULT_FULLY_IDLE_PULSE_INTERVAL;
	private volatile long cooldownInterval = DEFAULT_COOLDOWN_PEROID;
	private final long fullIdlePulseInterval = (1000 / 3);
	private volatile boolean __sig_stop_finished;

	public long getActivePulseInterval() {
		return this.activePulseInterval;
	}

	public long getIdlePulseInterval() {
		return this.idlePulseInterval;
	}

	public void setActivePulseInterval(final long interval) {
		if (interval <= 5) {
			return;
		} else {
			this.activePulseInterval = interval;
		}
	}

	public void setIdlePulseInterval(final long interval) {
		if (interval <= 5) {
			return;
		} else {
			this.idlePulseInterval = interval;
		}
	}

	@Override
	public void execute(final Runnable r) {
		TEConditions.validRunnable(r);
		this.run_task(new FutureTask<Object>() {

			@Override
			public Object execute() {
				r.run();
				return null;
			}

		});
	}

	private void __sig_term() {
		this.sig_stop.set(true);
		for (final PBTEThread dt : this.executors) {
			dt.stopThread();
			this.executors.remove(dt);
		}
		this.__sig_stop_finished = true;
		this.pbteController.interrupt();
	}

	@Override
	public boolean awaitTermination(final long amount, final TimeUnit unit) throws InterruptedException {
		this.__sig_term();
		return this.isTerminated();
	}

	@Override
	public boolean isShutdown() {
		return this.sig_stop.get();
	}

	@Override
	public boolean isTerminated() {
		return this.__sig_stop_finished;
	}

	@Override
	public void shutdown() {
		// Shuts down threads after their tasks finish.
		this.sig_stop.set(true);
	}

	@Override
	public List<Runnable> shutdownNow() {
		final BlockingQueue<Executable<?>> tasks = this.tasks; // .clone();
		// this.tasks.clear();
		this.__sig_term();
		final ArrayList<Runnable> converted = new ArrayList<Runnable>(tasks.size());
		for (final Executable<?> t : tasks) {
			converted.add(new Runnable() {

				@Override
				public void run() {
					t.run_st();
				}
			});
		}
		return converted;
	}

	private void run_task(final Executable<?> t) {
		try {
			this.tasks.transfer(t);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public String toString() {
		return "PBTE[threads=" + this.threadLimit + "]";
	}

	@SuppressWarnings("static-access")
	@Override
	public void awaitIdle() {
		final int limit = 8;
		int waitingInt = limit;
		final int s = (int) (35 / limit);
		while (!this.tasks.isEmpty()) {
			try {
				if (waitingInt <= 0) {
					waitingInt = limit + 1;
				}
				Thread.sleep((long) FastMath.get().ceil(s / waitingInt));
				waitingInt -= 1;
			} catch (InterruptedException e) {
			}
		}
	}

	private static final ExecutorFeatureExtensions[] SUPPORTED = new ExecutorFeatureExtensions[]
		{ ExecutorFeatureExtensions.EEX_executor_await_idle, ExecutorFeatureExtensions.EEX_query_batching_supported,
				ExecutorFeatureExtensions.EEX_query_slots_supported, };

	@Override
	public boolean supportsFeature(final ExecutorFeatureExtensions feature) {
		return ArrayUtils.contains(SUPPORTED, feature);
	}

	@Override
	public ExecutorFeatureExtensions[] supported() {
		return SUPPORTED;
	}

	/* EEX_query_batching_supported */

	@Override
	public boolean executorSupportsBatching() {
		return false;
	}

	/* EEX_query_slots_supported */

	@Override
	public boolean executorSupportsSlotCount() {
		return false;
	}
}
